//===----------------------------------------------------------------------===//
//
//                         BusTub
//
// update_executor.cpp
//
// Identification: src/execution/update_executor.cpp
//
// Copyright (c) 2015-2021, Carnegie Mellon University Database Group
//
//===----------------------------------------------------------------------===//
#include <memory>

#include "execution/executors/update_executor.h"

namespace bustub {

UpdateExecutor::UpdateExecutor(ExecutorContext *exec_ctx, const UpdatePlanNode *plan,
                               std::unique_ptr<AbstractExecutor> &&child_executor)
    : AbstractExecutor(exec_ctx)
    ,plan_(plan)
    ,child_executor_(std::move(child_executor)){}

void UpdateExecutor::Init() {
  table_info_ = exec_ctx_->GetCatalog()->GetTable(plan_->TableOid());
  child_executor_->Init();
}

/*
 * GenerateUpdatedTuple(const Tuple &src_tuple)函数用于更新一条记录
 * 我们只需要遍历子查询的结果然后调用该函数获得新记录并使用TableHeap中的UpdateTuple更新记录即可
 * 处理一下查询的错误，catch之后，往上抛
 * */
bool UpdateExecutor::Next([[maybe_unused]] Tuple *tuple, RID *rid) {
  Tuple old_tuple;
  Tuple new_tuple;
  RID tuple_rid;
  Transaction *transaction = GetExecutorContext()->GetTransaction();
  LockManager *lock_manager = GetExecutorContext()->GetLockManager();
  while(true){
    // 执行子查询
    try {
      if(!child_executor_->Next(&old_tuple,&tuple_rid)){
        break;
      }
    }catch (Exception &e){
      throw Exception(ExceptionType::UNKNOWN_TYPE,"updateExecutor::child execute error");
      return false;
    }

    // 加锁
    if(lock_manager != nullptr){
      if(transaction->IsSharedLocked(tuple_rid)){
        lock_manager->LockUpgrade(transaction,tuple_rid);
      }else if(!transaction->IsExclusiveLocked(tuple_rid)){
        lock_manager->LockExclusive(transaction,tuple_rid);
      }
    }


    // 更新记录-- 先是生成new_tuple然后将其插入到page中去。
    new_tuple = GenerateUpdatedTuple(old_tuple);
    TableHeap* table_heap = table_info_->table_.get();
    table_heap->UpdateTuple(new_tuple,tuple_rid,exec_ctx_->GetTransaction());

    // 还要更新索引
    for(const auto &index:exec_ctx_->GetCatalog()->GetTableIndexes(table_info_->name_)){
      // 先删除旧索引后增加新索引
      auto index_info = index->index_.get();
      index_info->DeleteEntry(
          old_tuple.KeyFromTuple(table_info_->schema_,*index_info->GetKeySchema(),index_info->GetKeyAttrs())
                                 ,tuple_rid,exec_ctx_->GetTransaction());
      index_info->InsertEntry(
          new_tuple.KeyFromTuple(table_info_->schema_,*index_info->GetKeySchema(),index_info->GetKeyAttrs()),
          tuple_rid,exec_ctx_->GetTransaction()
          );

      IndexWriteRecord write_record(tuple_rid,table_info_->oid_,WType::DELETE,
                                    new_tuple,index->index_oid_,exec_ctx_->GetCatalog());
      write_record.old_tuple_ = old_tuple;
      transaction->GetIndexWriteSet()->emplace_back(write_record);
    }

    // 解锁
    if(transaction->GetIsolationLevel() == IsolationLevel::READ_COMMITTED && lock_manager != nullptr){
      lock_manager->Unlock(transaction,tuple_rid);
    }
  }
  return false;
}


Tuple UpdateExecutor::GenerateUpdatedTuple(const Tuple &src_tuple) {
//  GetUpdateAtrr:更新属性
  const auto &update_attrs = plan_->GetUpdateAttr();
  Schema schema = table_info_->schema_;
  uint32_t col_count = schema.GetColumnCount();
  std::vector<Value> values;
  for (uint32_t idx = 0; idx < col_count; idx++) {
    if (update_attrs.find(idx) == update_attrs.cend()) {
      values.emplace_back(src_tuple.GetValue(&schema, idx));
    } else {
      const UpdateInfo info = update_attrs.at(idx);
      Value val = src_tuple.GetValue(&schema, idx);
      switch (info.type_) {
        case UpdateType::Add:
          values.emplace_back(val.Add(ValueFactory::GetIntegerValue(info.update_val_)));
          break;
        case UpdateType::Set:
          values.emplace_back(ValueFactory::GetIntegerValue(info.update_val_));
          break;
      }
    }
  }
  return Tuple{values, &schema};
}

}  // namespace bustub
